package com.chinamobile.iot.mqtt.schedule;

import com.chinamobile.iot.utils.NameThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 延迟执行某个任务时, 可以用此方法
 * 使用场景如下：
 * a、向对端发送某个消息时，发送失败，延迟多少秒以后再试
 * b、向对方发送消息成功后，把消息放入等待队列，在一段时间后检查是否收到响应
 * <p>
 * 一个DelayTask对象中的所有节点的超时时间应该是相同的，比如都是30秒等等，不支持随意设置超时时间
 * 上述限制主要是基于性能和实际的使用场景考虑
 * <p>
 * 在对象没超时之前，支持重复add设置新的超时时间，也可以remove。 当设置新的超时时间后，原来add设置时间到达时，不会触发超时。
 * <p>
 * Created by szl on 2017/6/28.
 */
public class DelayTask<T> {

    private static Logger logger = LoggerFactory.getLogger(DelayTask.class);
    /**
     * 超时队列，队列中的元素按照时间顺序排序，队列头部的时间最小，尾部的时间最大
     */
    private final ConcurrentLinkedQueue<Task<T>> queue = new ConcurrentLinkedQueue<>();

    /**
     * 所有待超时结点的hash表
     */
    private final ConcurrentHashMap<T, Long> map = new ConcurrentHashMap<>();

    /**
     * 负责进行超时检查的线程池
     */
    private final ScheduledExecutorService executorService;
    /**
     * 超时事件触发器函数
     */
    private TimerCallback<T> timerCallback;
    /**
     * 每个结点默认的超时时长，单位为秒
     */
    private final int timeout;
    /**
     * 检查的时间间隔,单位为秒
     */
    private final int checkRate;
    /**
     * 名字,也是线程的名字
     */
    private String name;

    public DelayTask(TimerCallback<T> timerCallback, int timeout) {
        this(timerCallback, timeout, 1, "delay-task-");
    }

    /**
     * Construct a delayTask
     *
     * @param timerCallback the timeout callback
     * @param timeout       timeOut  second
     * @param checkRate     check second
     * @param name          the background thread name
     */
    public DelayTask(TimerCallback<T> timerCallback, int timeout, int checkRate, String name) {
        this.timerCallback = timerCallback;
        this.timeout = timeout;
        this.checkRate = checkRate;
        this.name = name;
        this.executorService = Executors.newScheduledThreadPool(1, new NameThreadFactory(name));
    }

    public DelayTask(ScheduledExecutorService executorService, TimerCallback<T> timerCallback, int timeout, int checkRate) {
        this.executorService = executorService;
        this.timerCallback = timerCallback;
        this.timeout = timeout;
        this.checkRate = checkRate;
        this.name = "null";
    }

    public ScheduledExecutorService getExecutorService() {
        return executorService;
    }

    public TimerCallback<T> getTimerCallback() {
        return timerCallback;
    }

    public void setTimerCallback(TimerCallback<T> timerCallback) {
        this.timerCallback = timerCallback;
    }

    public int getTimeout() {
        return timeout;
    }

    public void start() {
        final DelayTask<T> delayTask = this;
        executorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                long now = System.currentTimeMillis();
                Task<T> task = queue.peek();
                while (task != null) {
                    if (task.dueTime <= now) {
                        Long dueTime = map.get(task.getObj());
                        if (dueTime != null) {
                            if (dueTime.equals(task.getDueTime())) {
                                map.remove(task.getObj());
                                if (timerCallback != null) {
                                    try {
                                        timerCallback.onTimer(task.dueTime, task.getObj(), delayTask);
                                    } catch (Exception e) {
                                        logger.error("error occured when execut the timerCallback::onTimer, {}", task.getObj().toString());
                                    }
                                }
                            }
                        }
                        // 把首节点移除
                        queue.poll();
                        // 从头部取出下一个结点
                        task = queue.peek();
                    } else {
                        break;
                    }
                }
            }
        }, 0, 1, TimeUnit.SECONDS);
    }

    public void stop() {
        executorService.shutdown();
    }

    public long add(T t) {
        long now = System.currentTimeMillis();
        long due = now + timeout * 1000;
        Task<T> task = new Task<>(due, t);
        queue.add(task);
        map.put(t, due);
        return due;
    }

    public void remove(T t) {
        map.remove(t);
    }

    /**
     * Created by szl on 2017/6/28.
     * <p>
     * DelayTask中的回调接口
     * <p>
     * 外部实现一个这种方法对象，注入到delayTask中，当某个元素超时时，会回调onTime方法
     */
    public interface TimerCallback<T> {
        void onTimer(long time, T t, DelayTask<T> task);
    }


    private static class Task<T> {
        /**
         * 超时的绝对时间
         */
        private long dueTime;
        /**
         * 超时的对象
         */
        private T t;

        public Task() {
        }

        private Task(long dueTime, T t) {
            this.dueTime = dueTime;
            this.t = t;
        }

        private long getDueTime() {
            return dueTime;
        }

        public void setDueTime(long dueTime) {
            this.dueTime = dueTime;
        }

        private T getObj() {
            return t;
        }

        private void setObj(T t) {
            this.t = t;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) return true;
            if (o == null || getClass() != o.getClass()) return false;

            Task<?> task = (Task<?>) o;

            return dueTime == task.dueTime && t.equals(task.t);
        }

        @Override
        public int hashCode() {
            int result = (int) (dueTime ^ (dueTime >>> 32));
            result = 31 * result + t.hashCode();
            return result;
        }
    }

    private static class MyThreadFactory implements ThreadFactory {
        private final AtomicInteger count = new AtomicInteger(0);
        private final String baseName;

        private MyThreadFactory(String baseName) {
            this.baseName = baseName;
        }

        @Override
        public Thread newThread(Runnable r) {
            int number = count.getAndIncrement();
            return new Thread(r, baseName + number);
        }
    }
}
